#include "xlcomm/net/tcp_connection.h"

#include "xlcomm/base/logger.h"
#include "xlcomm/net/detail/channel.h"
#include "xlcomm/net/detail/socket.h"
#include "xlcomm/net/detail/sockets_ops.h"

namespace xlcomm {
namespace net {

TcpConnection::TcpConnection(EventLoop* loop, const std::string& name,
                             int sockfd, const InetAddress& local_addr,
                             const InetAddress& peer_addr)
    : loop_(loop),
      name_(name),
      socket_(std::make_unique<Socket>(sockfd)),
      channel_(std::make_unique<Channel>(loop, sockfd)),
      local_addr_(local_addr),
      peer_addr_(peer_addr),
      state_(kConnecting),
      context_(nullptr),
      high_water_size_(0) {
  LOG_DEBUG << "TcpConnection(conn_fd:" << sockfd << ") " << name_
            << " created, addr:" << this;
  XLCOMM_ASSERT(loop_ != nullptr);
  channel_->set_read_callback(std::bind(&TcpConnection::HandleRead, this, _1));
  channel_->set_write_callback(std::bind(&TcpConnection::HandleWrite, this));
  channel_->set_close_callback(std::bind(&TcpConnection::HandleClose, this));
  channel_->set_error_callback(std::bind(&TcpConnection::HandleError, this));
}

TcpConnection::~TcpConnection() {
  LOG_DEBUG << "TcpConnection(conn_fd:" << channel_->fd() << ") " << name_
            << " dtor. Auto close fd";
  XLCOMM_ASSERT(state_ == kDisconnected);
}

void TcpConnection::ConnectEstablished() {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  XLCOMM_ASSERT(state_ == kConnecting);
  state_ = kConnected;
  channel_->Tie(shared_from_this());
  channel_->EnableReading();
  if (connect_callback_) {
    connect_callback_(shared_from_this());
  }
}

// 这是TcpConnection析构前调用的最后一个成员函数
void TcpConnection::ConnectDestroyed() {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  if (state_ == kConnected) {
    state_ = kDisconnected;
    channel_->DisableAll();
    if (disconnect_callback_) {
      disconnect_callback_(shared_from_this());
    }
  }
  channel_->Remove();
}

void TcpConnection::HandleRead(Timestamp recv_time) {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  int saved_errno = 0;
  ssize_t n = input_buffer_.ReadFromFd(channel_->fd(), &saved_errno);
  LOG_DEBUG << "TcpConnection(conn_fd:" << channel_->fd() << ") read some";
  if (n > 0) {
    if (message_callback_) {
      message_callback_(shared_from_this(), &input_buffer_, recv_time);
    }
  } else if (n == 0) {
    LOG_INFO << "read 0, peer close connection " << name_;
    HandleClose();
  } else {
    errno = saved_errno;
    LOG_ERROR << "TcpConnection " << name_ << " HandleRead error";
    HandleError();
  }
}

void TcpConnection::HandleClose() {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  LOG_DEBUG << "TcpConnection(conn_fd:" << channel_->fd() << ") will close";
  XLCOMM_ASSERT(state_ == kConnected || state_ == kDisconnecting);
  // we don't close fd, leave it to dtor, so we can find leaks easily.
  state_ = kDisconnected;
  channel_->DisableAll();
  if (disconnect_callback_) {
    disconnect_callback_(shared_from_this());
  }
  if (close_callback_) {
    // 必须在最后一行
    close_callback_(shared_from_this());
  }
}

void TcpConnection::HandleError() {
  int err = sockets::GetSocketError(channel_->fd());
  LOG_WARN << "TcpConnection " << name_ << " HandleError, only show SO_ERROR("
           << err << "):" << strerror_tl(err);
}

void TcpConnection::Send(const void* data, size_t len) {
  Send(std::string(static_cast<const char*>(data), len));
}

void TcpConnection::Send(const char* str) { Send(std::string(str)); }

void TcpConnection::Send(const std::string& message) {
  if (state_ == kConnected) {
    if (loop_->IsInLoopThread()) {
      SendInLoop(message);
    } else {
      void (TcpConnection::*fp)(const std::string& message) =
          &TcpConnection::SendInLoop;
      // 可优化：使用C++11的移动语义，避免内存拷贝的开销
      loop_->RunInLoop(std::bind(fp, shared_from_this(), message));
    }
  } else {
    LOG_WARN << "TcpConnection(conn_fd:" << channel_->fd()
             << ") is not connected, give up writing";
  }
}

void TcpConnection::Send(Buffer* buff) {
  if (state_ == kConnected) {
    if (loop_->IsInLoopThread()) {
      SendInLoop(buff->ReaderPtr(), buff->ReadableBytes());
      buff->HasReadAll();
    } else {
      void (TcpConnection::*fp)(const std::string& message) =
          &TcpConnection::SendInLoop;
      // 可优化：
      loop_->RunInLoop(
          std::bind(fp, shared_from_this(), buff->ReadAllAsString()));
    }
  } else {
    LOG_WARN << "TcpConnection(conn_fd:" << channel_->fd()
             << ") is not connected, give up writing";
  }
}

void TcpConnection::SendInLoop(const std::string& message) {
  SendInLoop(message.data(), message.size());
}

void TcpConnection::SendInLoop(const void* data, size_t len) {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  if (state_ == kDisconnected) {
    LOG_WARN << "TcpConnection(conn_fd:" << channel_->fd()
             << ") disconnected, give up writing";
    return;
  }

  bool faultError = false;
  ssize_t nwrote = 0;
  size_t remaining = len;
  // 没在监听可写事件、且输出缓冲区没数据，直接写socket
  if (!channel_->IsEnableWriting() && output_buffer_.ReadableBytes() == 0) {
    nwrote = sockets::write(channel_->fd(), data, len);
    if (nwrote >= 0) {
      remaining = len - nwrote;
      if (remaining == 0 && write_complete_callback_) {
        // QueueInLoop有延迟，RunInLoop可能阻塞当前执行 且 容易和Send循环调用
        loop_->QueueInLoop(
            std::bind(write_complete_callback_, shared_from_this()));
      }
    } else {  // nwrote < 0
      nwrote = 0;
      if (errno != EWOULDBLOCK) {
        LOG_SYSERROR(write) << "fd:" << channel_->fd();
        // signal(SIGPIPE, SIG_IGN)后不再产生SIGPIPE信号
        // 此后::write会返回-1，errno的值为EPIPE
        if (errno == EPIPE || errno == ECONNRESET)  // FIXME: any others?
        {
          faultError = true;
        }
      }
    }
  }

  // 没写完，放到output_buffer_并监听可写事件
  if (!faultError && remaining > 0) {
    size_t old_len = output_buffer_.ReadableBytes();
    if (old_len + remaining >= high_water_size_ && old_len < high_water_size_ &&
        outbuf_high_water_callback_) {
      // 只在上升沿触发一次
      loop_->QueueInLoop(std::bind(outbuf_high_water_callback_,
                                   shared_from_this(), old_len + remaining));
    }
    output_buffer_.Append(static_cast<const char*>(data) + nwrote, remaining);
    if (!channel_->IsEnableWriting()) channel_->EnableWriting();
  }
}

void TcpConnection::HandleWrite() {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  if (channel_->IsEnableWriting()) {
    ssize_t n = sockets::write(channel_->fd(), output_buffer_.ReaderPtr(),
                               output_buffer_.ReadableBytes());
    if (n > 0) {
      output_buffer_.HasRead(n);
      if (output_buffer_.ReadableBytes() == 0) {  // 已写完
        // 取消监听可写事件，避免进入busy-loop
        channel_->DisableWriting();
        if (write_complete_callback_) {
          // QueueInLoop有延迟，RunInLoop可能阻塞当前执行 且 容易和Send循环调用
          loop_->QueueInLoop(
              std::bind(write_complete_callback_, shared_from_this()));
        }
        // 用户调过Shutdown，写完后再shutdown
        if (state_ == kDisconnecting) {
          ShutdownInLoop();
        }
      }
    } else {
      LOG_SYSERROR(write) << "fd:" << channel_->fd();
    }
  } else {
    LOG_DEBUG << "Connection(conn_fd:" << channel_->fd()
              << ") is down, no more writing";
  }
}

void TcpConnection::Shutdown() {
  // FIXME: use compare and swap
  if (state_ == kConnected) {
    state_ = kDisconnecting;
    loop_->RunInLoop(
        std::bind(&TcpConnection::ShutdownInLoop, shared_from_this()));
  }
}

void TcpConnection::ShutdownInLoop() {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  if (!channel_->IsEnableWriting()) {  // 缓冲区没待写数据
    LOG_INFO << "local shutdown write " << name_;
    socket_->ShutdownWrite();
  }
}

void TcpConnection::ForceClose() {
  // FIXME: use compare and swap
  if (state_ == kConnected || state_ == kDisconnecting) {
    state_ = kDisconnecting;
    loop_->QueueInLoop(
        std::bind(&TcpConnection::ForceCloseInLoop, shared_from_this()));
  }
}

void TcpConnection::ForceCloseInLoop() {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  if (state_ == kConnected || state_ == kDisconnecting) {
    LOG_INFO << "local force close connection " << name_;
    HandleClose();
  }
}

void TcpConnection::SetTcpNoDelay(bool on) { socket_->SetTcpNoDelay(on); }

void TcpConnection::SetKeepAlive(bool on) { socket_->SetKeepAlive(on); }

const char* TcpConnection::StateToString() const {
  switch (state_) {
    case kConnecting:
      return "kConnecting";
    case kConnected:
      return "kConnected";
    case kDisconnecting:
      return "kDisconnecting";
    case kDisconnected:
      return "kDisconnected";
    default:
      return "unknown state";
  }
}

}  // namespace net
}  // namespace xlcomm